package com.zpa.redisstreamlistener.processor;

import com.alibaba.fastjson.JSONObject;
import com.zpa.redisstreamlistener.annotation.RedisStreamListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RedissonClient;
import org.redisson.api.StreamMessageId;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.validation.constraints.NotNull;
import java.lang.reflect.Method;
import java.util.Map;

/**
 * RedisStreamListener 注解处理器
 */
@Component
@Slf4j
public class RedisStreamListenerProcessor implements BeanPostProcessor, ApplicationContextAware {

    public static ApplicationContext applicationContext;

    @Override
    public void setApplicationContext(@NotNull ApplicationContext applicationContext) throws BeansException {
        RedisStreamListenerProcessor.applicationContext = applicationContext;
    }

    @Override
    public Object postProcessAfterInitialization(Object bean, @NotNull String beanName) throws BeansException {
        // 获取所有方法
        Class<?> targetClass = bean.getClass();
        Method[] methods = targetClass.getDeclaredMethods();
        // 遍历所有方法
        for (Method method : methods) {
            // 判断是否是RedisStreamListener注解
            if (method.isAnnotationPresent(RedisStreamListener.class)) {
                // 获取注解
                RedisStreamListener annotation = method.getAnnotation(RedisStreamListener.class);
                try {
                    // 启动监听器
                    startConsumerWithHandler(bean, method, annotation);
                } catch (Exception e) {
                    log.error("自动启动监听器失败", e);
                    throw new RuntimeException(e);
                }
            }
        }
        return bean;
    }


    /**
     * 启动监听器
     * @param bean bean的类
     * @param method bean其中之一的方法 因为调用这个方法的方法遍历了 bean 的所有方法
     * @param annotation 方法上的注解
     */
    private void startConsumerWithHandler(Object bean, Method method, RedisStreamListener annotation) {
        // 获取注解参数
        String topic = annotation.topic();
        String groupId = annotation.groupId();
        String factory = annotation.containerFactory();
        boolean autoSubmit = annotation.autoSubmit();
        // 获取 RedissonClient 注解中的 containerFactory 参数可以自定义 RedissonClient 的注入，测试用例有写
        // 使用默认 就是使用主要的 RedissonClient 的 Bean （@Bean + @Primary）
        RedissonClient redissonClient;
        if (StringUtils.isNotBlank(factory)) {
            redissonClient = applicationContext.getBean(factory, RedissonClient.class);
        } else {
            redissonClient = applicationContext.getBean(RedissonClient.class);
        }
        // 创建 RedisStreamConsumer 对象
        RedisStreamConsumer redisStreamConsumer = new RedisStreamConsumer(redissonClient);
        // 启动监听器
        redisStreamConsumer.startProcessing(topic, groupId, autoSubmit, (map, ack) -> handleStreamMessage(bean, method, map, ack));
    }

    /**
     * 处理消息
     * 主要用于赋值 并执行方法体
     */
    private void handleStreamMessage(Object bean, Method method, Map.Entry<StreamMessageId, Map<String, Object>> messageMap, Acknowledgment ack) {
        try {
            method.setAccessible(true);
            // 获取 方法 的所有参数类型
            Class<?>[] parameterTypes = method.getParameterTypes();
            // 根据方法参数类型传递合适的参数
            Object[] args = new Object[parameterTypes.length];
            for (int i = 0; i < parameterTypes.length; i++) {
                Class<?> parameterType = parameterTypes[i];
                // 获取 值
                if (parameterType.equals(Map.class)) {
                    Map<String, Object> value = messageMap.getValue();
                    args[i] = value;
                }
                // 序列化
                if (parameterType.equals(String.class)) {
                    Map<String, Object> value = messageMap.getValue();
                    args[i] = JSONObject.toJSONString(value);
                }
                // 获取 id
                if (StreamMessageId.class.isAssignableFrom(parameterType)) {
                    StreamMessageId key = messageMap.getKey();
                    args[i] = key;
                }
                // 手动提交配置
                if (Acknowledgment.class.isAssignableFrom(parameterType)) {
                    args[i] = ack;
                }
            }
            // 执行方法
            // 就是通过方式 把参数传递给方法体
            method.invoke(bean, args);
        } catch (Exception e) {
            log.error("handleStreamMessage 处理消息失败: {}", e.getMessage(), e);
            throw new RuntimeException(e);
        }
    }
}